package com.linkgie.galaxyframework.kvstorage;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.function.Function;
import java.util.function.Predicate;

import com.linkgie.galaxyframework.utils.Functions;
import com.linkgie.galaxyframework.utils.Predicates;

/**
 * 分页迭代器；
 * 
 * @author huanghaiquan
 *
 * @param <K>
 * @param <V>
 * @param <R>
 */
public class KeyPaginator<K, V, R> implements Iterator<Collection<K>> {

	public static final int DEFAULT_PAGE_SIZE = 10;

	private KeyIterator<K, V> iterator;

	private int pageSize;

	private Function<KVEntry<K, V>, R> converter;

	private Predicate<K> predicator;

	private KeyPaginator(int pageSize, KeyIterator<K, V> iterator, Predicate<K> predicator,
			Function<KVEntry<K, V>, R> converter) {
		this.pageSize = pageSize;
		this.iterator = iterator;
		this.predicator = predicator;
		this.converter = converter;
	}

	public int pageSize() {
		return pageSize;
	}

	@Override
	public boolean hasNext() {
		return iterator.hasNext();
	}

	/**
	 * 下一个键分页；
	 * <p>
	 * 最大长度为 {@link #pageSize()} ；
	 * <p>
	 * 
	 * 如果没有数据项，则返回 null；
	 */
	@Override
	public Collection<K> next() {
		List<K> keys = new ArrayList<K>();
		K k = null;
		int i = 0;
		while (i < pageSize && (k = iterator.next()) != null) {
			if (predicator.test(k)) {
				keys.add(k);
				i++;
			}
		}
		if (i == 0) {
			return null;
		}
		return Collections.unmodifiableCollection(keys);
	}

	/**
	 * 下一个值分页；
	 * <p>
	 * 最大长度为 {@link #pageSize()} ；
	 * <p>
	 * 
	 * 如果没有数据项，则返回 null；
	 * 
	 * @return
	 */
	public Collection<R> nextPage() {
		List<R> values = new ArrayList<R>();
		KVEntry<K, V> e = null;
		R r = null;
		int i = 0;
		while (i < pageSize && (e = iterator.nextEntry()) != null) {
			if (predicator.test(e.getKey())) {
				r = converter.apply(e);
				if (r != null) {
					values.add(r);
					i++;
				}
			}
		}
		if (i == 0) {
			return null;
		}
		return Collections.unmodifiableCollection(values);
	}

	/**
	 * 下一个键值项分页；
	 * <p>
	 * 最大长度为 {@link #pageSize()} ；
	 * <p>
	 * 
	 * 如果没有数据项，则返回 null；
	 * 
	 * @return
	 */
	public Collection<KVEntry<K, R>> nextPageEntries() {
		List<KVEntry<K, R>> entries = new ArrayList<KVEntry<K, R>>();
		KVEntry<K, V> e = null;
		R r = null;
		int i = 0;
		while (i < pageSize && (e = iterator.nextEntry()) != null) {
			if (predicator.test(e.getKey())) {
				r = converter.apply(e);
				if (r != null) {
					entries.add(new Entry<K, R>(e.getKey(), r));
					i++;
				}
			}
		}
		if (i == 0) {
			return null;
		}
		return Collections.unmodifiableCollection(entries);
	}

	public static <T> PaginationBuilder<T> iterator(KVMap<T> kvMap) {
		return new PaginationBuilder<>(kvMap);
	}

	// ----------------- type ------------------

	/**
	 * 分页检索构建器；
	 * 
	 * @author huanghaiquan
	 *
	 * @param <V> 值的类型；
	 */
	public static class PaginationBuilder<V> {

		private KVMap<V> kvMap;

		private int pageSize = DEFAULT_PAGE_SIZE;

		private PaginationBuilder(KVMap<V> kvMap) {
			this.kvMap = kvMap;
		}

		public PaginationBuilder<V> pageSize(int pageSize) {
			if (pageSize < 1 || pageSize > 10000) {
				throw new IllegalArgumentException("The page size is out of bound [1, 10000]!");
			}
			this.pageSize = pageSize;
			return this;
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * 返回结果不包含指定的起始键；
		 * <p>
		 * 
		 * @param key
		 * @return
		 */
		public ScopedBuilder<String, V> startKeyExcluded(String key) {
			return startKey(key, true);
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * @param key      起始键；
		 * @param excluded 是否排除起始键；
		 * @return
		 */
		public ScopedBuilder<String, V> startKey(String key, boolean excluded) {
			ScopedBuilder<String, V> builder = new StringKeyScopedBuilder<V>(this);
			builder.startKey(key, excluded);

			return builder;
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * 返回结果不包含指定的起始键；
		 * <p>
		 * 
		 * @param key
		 * @return
		 */
		public ScopedBuilder<byte[], V> startKeyExcluded(byte[] key) {
			return startKey(key, true);
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * @param key      起始键；
		 * @param excluded 是否排除起始键；
		 * @return
		 */
		public ScopedBuilder<byte[], V> startKey(byte[] key, boolean excluded) {
			ScopedBuilder<byte[], V> builder = new BytesKeyScopedBuilder<V>(this);
			builder.startKey(key, excluded);

			return builder;
		}

		/**
		 * 截止键；
		 * <p>
		 * 
		 * 返回结果不包含指定的截止键；
		 * <p>
		 * 
		 * @param key 截止键；
		 * @return
		 */
		public ScopedBuilder<String, V> endKeyExcluded(String key) {
			return endKey(key, true);
		}

		/**
		 * 截止键；
		 * <p>
		 * 
		 * @param key      截止键；
		 * @param excluded 是否排除截止键；
		 * @return
		 */
		public ScopedBuilder<String, V> endKey(String key, boolean excluded) {
			ScopedBuilder<String, V> builder = new StringKeyScopedBuilder<V>(this);
			builder.endKey(key, excluded);

			return builder;
		}

		/**
		 * 截止键；
		 * <p>
		 * 
		 * 返回结果不包含指定的截止键；
		 * <p>
		 * 
		 * @param key 截止键；
		 * @return
		 */
		public ScopedBuilder<byte[], V> endKeyExcluded(byte[] key) {
			return endKey(key, true);
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * @param key      起始键；
		 * @param excluded 是否排除起始键；
		 * @return
		 */
		public ScopedBuilder<byte[], V> endKey(byte[] key, boolean excluded) {
			ScopedBuilder<byte[], V> builder = new BytesKeyScopedBuilder<V>(this);
			builder.endKey(key, excluded);

			return builder;
		}

	}

	/**
	 * 带范围的分页检索构建器；
	 * 
	 * @author huanghaiquan
	 *
	 * @param <K>
	 * @param <V>
	 */
	public static abstract class ScopedBuilder<K, V> {

		private PaginationBuilder<V> pagination;

		private K startKey;

		private K endKey;

		private boolean excludedStart;

		private boolean excludedEnd;
		
		private boolean asc = true;

		private ScopedBuilder(PaginationBuilder<V> pagination) {
			this.pagination = pagination;
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * 返回结果不包含指定的起始键；
		 * <p>
		 * 
		 * @param key
		 * @return
		 */
		public ScopedBuilder<K, V> startKeyExcluded(K key) {
			return startKey(key, true);
		}

		/**
		 * 起始键；
		 * <p>
		 * 
		 * @param key      起始键；
		 * @param excluded 是否排除起始键；
		 * @return
		 */
		public ScopedBuilder<K, V> startKey(K key, boolean excluded) {
			this.startKey = key;
			this.excludedStart = excluded;

			return this;
		}

		/**
		 * 排除终点；
		 * 
		 * @param key 截止键；结果不包含指定的键；
		 * @return
		 */
		public ScopedBuilder<K, V> endKeyExcluded(K key) {
			return endKey(key, true);
		}

		/**
		 * 终点；
		 * @param key
		 * @param excluded
		 * @return
		 */
		public ScopedBuilder<K, V> endKey(K key, boolean excluded) {
			this.endKey = key;
			this.excludedEnd = excluded;

			return this;
		}
		
		public ScopedBuilder<K, V> asc(){
			this.asc = true;
			return this;
		}
		
		public ScopedBuilder<K, V> desc(){
			this.asc = false;
			return this;
		}
		
		public ScopedBuilder<K, V> pageSize(int pageSize){
			this.pagination.pageSize(pageSize);
			return this;
		}

		abstract KeyIterator<K, V> seek(KVMap<V> kvMap, K startKey, boolean excludedStart, K endKey,
				boolean excludedEnd, boolean asc);

		/**
		 * 创建分页迭代器；
		 * 
		 * @return
		 */
		public KeyPaginator<K, V, V> build() {
			return build(Predicates.alwaysTrue(), Functions.constant());
		}

		/**
		 * 创建分页迭代器；
		 * 
		 * @param filter 键的过滤器；只有通过过滤器检测的键才会包含在结果中；
		 * @return
		 */
		public KeyPaginator<K, V, V> build(Predicate<K> filter) {
			return build(filter, Functions.constant());
		}

		/**
		 * 创建分页迭代器；
		 * 
		 * @param <R>       输出类型；
		 * @param converter 转换器；将值转为输出类型；输出 null 的数据项会被忽略；
		 * @return
		 */
		public <R> KeyPaginator<K, V, R> build(Function<KVEntry<K, V>, R> converter) {
			return build(Predicates.alwaysTrue(), converter);
		}

		/**
		 * 创建分页迭代器；
		 * 
		 * @param <R>       输出类型；
		 * @param filter    键的过滤器；只有通过过滤器检测的键才会包含在结果中；
		 * @param converter 转换器；将值转为输出类型；输出 null 的数据项会被忽略；
		 * @return
		 */
		public <R> KeyPaginator<K, V, R> build(Predicate<K> filter, Function<KVEntry<K, V>, R> converter) {
			KeyIterator<K, V> iterator = seek(pagination.kvMap, startKey, excludedStart, endKey, excludedEnd, asc);
			return new KeyPaginator<K, V, R>(pagination.pageSize, iterator, filter, converter);
		}

	}

	private static class StringKeyScopedBuilder<V> extends ScopedBuilder<String, V> {

		public StringKeyScopedBuilder(PaginationBuilder<V> pagination) {
			super(pagination);
		}

		@Override
		KeyIterator<String, V> seek(KVMap<V> kvMap, String startKey, boolean excludedStart, String endKey,
				boolean excludedEnd, boolean asc) {
			if (asc) {
				return kvMap.prefixSeek(startKey, excludedStart, endKey, excludedEnd);
			}
			return kvMap.prefixSeekDesc(startKey, excludedStart, endKey, excludedEnd);
		}
	}

	private static class BytesKeyScopedBuilder<V> extends ScopedBuilder<byte[], V> {

		public BytesKeyScopedBuilder(PaginationBuilder<V> pagination) {
			super(pagination);
		}

		@Override
		KeyIterator<byte[], V> seek(KVMap<V> kvMap, byte[] startKey, boolean excludedStart, byte[] endKey,
				boolean excludedEnd, boolean asc) {
			if (asc) {
				return kvMap.prefixSeek(startKey, excludedStart, endKey, excludedEnd);
			}
			return kvMap.prefixSeekDesc(startKey, excludedStart, endKey, excludedEnd);
		}

	}

	private static class Entry<K, V> implements KVEntry<K, V> {

		private K key;

		private V value;

		public Entry(K key, V value) {
			this.key = key;
			this.value = value;
		}

		@Override
		public K getKey() {
			return key;
		}

		@Override
		public V getValue() {
			return value;
		}

	}

}
